#include "ZkClient.h"
#include "MyRpc.h"
#include <iostream>
#include <semaphore.h>
#include <string>

// 全局的watcher  响应zkserver给zkclient的通知
void global_watcher(zhandle_t* zh, int type, int state, const char* path, void* watcher_ctx) 
{
    // 会话事件
    if (type == ZOO_SESSION_EVENT) 
    {
        // 连接成功
        if (state == ZOO_CONNECTED_STATE) 
        {
            sem_t* sem = (sem_t*)zoo_get_context(zh);
            sem_post(sem);
        }
    }
}

ZkClient::ZkClient(): m_zk_handle_(nullptr) {}
ZkClient::~ZkClient() 
{
    if (m_zk_handle_) {
        zookeeper_close(m_zk_handle_);
    }
}

// 连接 zk 服务端
void ZkClient::Connect(std::string ip, uint16_t port) 
{
    // 连接zk服务端
    // zookeeper会创建两个新线程  
    // 网络IO线程负责连接服务端  watcher线程负责处理zkserver给zkclient的通知
    std::string host_port = ip + ":" + std::to_string(port);
    m_zk_handle_ = zookeeper_init(host_port.c_str(), global_watcher, 30000, nullptr, nullptr, 0);

    if (m_zk_handle_ == nullptr) 
    {
        LOG_ERROR("zookeeper_init failed");
        // 可能是内存不足等严重问题 退出进程
        exit(EXIT_FAILURE);
    }

    // 使用信号量来阻塞当前线程 等待连接服务端成功
    sem_t sem;
    sem_init(&sem, 0, 0);
    zoo_set_context(m_zk_handle_, &sem);
    sem_wait(&sem);

    LOG_INFO("zookeeper_init success");
}
// 创建 znode 指定路径 数据 是否临时节点
bool ZkClient::CreateNode(std::string path, std::string data, bool is_ephemeral) 
{
    // 设置 flag
    int flag = 0;
    if (is_ephemeral) 
    {
        flag = flag | ZOO_EPHEMERAL;
    }

    // 创建 znode
    int ret = zoo_create(m_zk_handle_, path.c_str(), data.c_str(), data.size(), &ZOO_OPEN_ACL_UNSAFE, flag, nullptr, 0);
    if (ret == ZOK)
    {
        LOG_INFO("znode create success: %s ->  %s", path.c_str(), data.c_str());
        return true;
    }
    else if(ret == ZNODEEXISTS)
    {
        LOG_INFO("znode exists: %s", path.c_str());
        return true;
    }
    else
    {
        LOG_ERROR("zoo_create failed, ret: %d", ret);
        return false;
    }
}
// 删除 znode 指定路径
bool ZkClient::DeleteNode(std::string path) 
{
    // 删除 znode
    int ret = zoo_delete(m_zk_handle_, path.c_str(), -1);
    if (ret != ZOK) 
    {
        LOG_ERROR("zoo_delete failed, ret: %d", ret);
        return false;
    }
    LOG_INFO("znode delete success: %s", path.c_str());
    return true;
}
// 获取数据 指定路径
std::string ZkClient::GetData(std::string path) 
{
    // 获取数据
    char buffer[64];
    int buffer_len = sizeof(buffer);
    int ret = zoo_get(m_zk_handle_, path.c_str(), 0, buffer, &buffer_len, nullptr);
    if (ret != ZOK) 
    {
        LOG_ERROR("zoo_get failed, ret: %d", ret);
        return "";
    }
    LOG_INFO("znode get success: %s -> %s", path.c_str(), buffer);
    return std::string(buffer);
}
// 设置数据 指定路径 数据
bool ZkClient::SetData(std::string path, std::string data)
{
    // 设置数据
    int ret = zoo_set(m_zk_handle_, path.c_str(), data.c_str(), data.size(), -1);
    if (ret != ZOK) 
    {
        LOG_ERROR("zoo_set failed, ret: %d", ret);
        return false;
    }
    LOG_INFO("znode set success: %s -> %s", path.c_str(), data.c_str());
    return true;
}
